package com.springstack.kafka.producer.config;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.HashMap;

/**
 * @author qiangfanghao
 * @date 2022/7/21
 * KafkaProducer配置类
 */
@Slf4j
@Configuration
@EnableConfigurationProperties(KafkaProducerProperties.class)
public class KafkaProducerConfig {

    @Bean(name = "kafkaTemplate")
    public KafkaTemplate<String,String> kafkaTemplate(KafkaProducerProperties kafkaProducerProperties) {
        HashMap<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProducerProperties.getBootstrapServers());
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,kafkaProducerProperties.getKeySerializer());
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,kafkaProducerProperties.getValueSerializer());
        configs.put(ProducerConfig.ACKS_CONFIG,kafkaProducerProperties.getAcks());
        configs.put(ProducerConfig.BATCH_SIZE_CONFIG,kafkaProducerProperties.getBatchSize());
        configs.put(ProducerConfig.BUFFER_MEMORY_CONFIG,kafkaProducerProperties.getBufferMemory());
        configs.put(ProducerConfig.COMPRESSION_TYPE_CONFIG,kafkaProducerProperties.getCompressionType());
        configs.put(ProducerConfig.LINGER_MS_CONFIG,kafkaProducerProperties.getLingerMs());
        DefaultKafkaProducerFactory<String,String> producerFactory = new DefaultKafkaProducerFactory<String,String>(configs);
        log.debug("kafkaTemplate config properties -> {}",kafkaProducerProperties.toString());
        return new KafkaTemplate<String,String>(producerFactory);
    }
}
